# encoding=utf8
# 这里的函数应该是可以跨应用的，最好无依赖
import base64
import io
import tempfile
import time
from typing import *
from typing import BinaryIO

__all__ = [
    "line_2_map",
    "map_2_line",
    "time_now_str",
    "base64_encode",
    "base64_decode",
    "base64_decode_image_fp",
    "base64_decode_video_fp",
    "base64_encode_image_fp",
    # "open_fp_from_url",
    "hexs_to_bytes",
    "bytes_to_hexs",
    "vec_to_string",
    "close_client",
    "strdecode",
    "parse_bool",
    # "str_similarity"
    "group_every_size",
    "read_all"
]


# 读文本文件内容到
def read_all(file_path: str) -> str:
    with open(file_path, "r") as f:
        return str(f.read())


def group_every_size(ss: List[str], every: int = 3) -> List[List[str]]:
    rs = []
    curr = []
    for i, s in enumerate(ss):
        if i != 0 and i % every == 0:
            rs.append(curr)
            curr = []
        curr.append(s)
    if len(curr) > 0:
        rs.append(curr)
    return rs


def line_2_map(line: str, seq1: str = "`", seq2: str = "="):
    m = {}
    fs = line.split(seq1)
    for f in fs:
        ss = f.split(seq2, 1)
        k = ss[0]
        if len(ss) >= 2:
            v = ss[1]
        else:
            v = ""
        m[k] = v
    return m


def map_2_line(m: Dict[str, any], seq1: str = "`", seq2: str = "="):
    fs = []
    for k, v in m.items():
        f = f"{k}{seq2}{v}"
        fs.append(f)
    return seq1.join(fs)


def time_now_str():
    return time.strftime("%Y%m%d_%H%M%S")


def md5_bs(bs: bytearray) -> str:
    import hashlib
    m = hashlib.md5()
    m.update(bs)
    return m.hexdigest()


def md5(s: str) -> str:
    bs = s.encode("utf8")
    return md5_bs(bs)


def md5_fp(fp: BinaryIO) -> str:
    fp.seek(0)
    bs = fp.read()
    return md5_bs(bs)


# 你好 -> 5L2g5aW9
def base64_encode(string):
    bs = string.encode('utf-8')
    bs = base64.standard_b64encode(bs)
    s = str(bs, 'utf-8')
    return s


# 5L2g5aW9 -> 你好
def base64_decode(string):
    bs = string.encode('utf-8')
    bs = base64.standard_b64decode(bs)
    s = str(bs, 'utf-8')
    return s


def base64_decode_bytearray(string):
    bs = string.encode('utf-8')
    bs = base64.standard_b64decode(bs)
    return bs


def base64_encode_image_fp(img_fp):
    img_fp.seek(0)
    bs = img_fp.read()
    bs = base64.standard_b64encode(bs)
    s = str(bs, 'utf-8')
    return s


def base64_decode_image_fp(string):
    bs = string.encode('utf-8')
    bs = base64.standard_b64decode(bs)
    fd = io.BytesIO(bs)
    return fd


def base64_decode_video_fp(string):
    """
    需要落盘，需要ffmpeg
    :param string:
    :return: 下载的文件名
    """
    bs = string.encode('utf-8')
    bs = base64.standard_b64decode(bs)

    tmp_dir_path = tempfile.gettempdir()
    tmp_file_path = "%s/vid_hash_%d.mp4" % (tmp_dir_path, hash(string))
    with open(tmp_file_path, "wb") as f:
        f.write(bs)
    return tmp_file_path


# def open_fp_from_url(url) -> Optional[io.BytesIO]:
#     """
#     打开一个url,file并构建fd对象，此对象可以在pil open中打开
#     :param url:
#     :return:
#     """
#     if url.startswith("file://"):
#         file_path = url[7:]
#         with open(file_path, "rb") as f:
#             fd = io.BytesIO(f.read())
#             return fd
#     r = requests.get(url, timeout=15, stream=True)
#     if r.status_code == 200:
#         fd = io.BytesIO(r.content)
#         return fd
#     return None


def hexs_to_bytes(hexs):
    """
    :param hexs: 16进制的字符串  ABCDEF0123456789AB
    :return: bytes 对象 py3
    """
    return bytes.fromhex(hexs)


def bytes_to_hexs(bs):
    return ''.join(['%02X' % b for b in bs])


def vec_to_string(vec):
    """
    把向量字符化
    :param vec: 向量数组 [float]
    :return: string
    """
    ss = []
    _min = 0.00000001
    for f in vec:
        if f < _min:
            s = "0"
        else:
            s = "%.8f" % f
            s = s[1:]
        ss.append(s)
    return ",".join(ss)


def close_client(client):
    """
    安静的关闭，调用close函数，如果有的话
    :param client:
    """
    if client is not None:
        try:
            if hasattr(client, 'close'):
                client.close()
        except Exception:
            pass


def parse_bool(s):
    s = str(s).lower().strip()
    false_ks = {'false', '0', 'no', ''}
    if s in false_ks:
        return False
    return True


# 仅用于py3
def strdecode(sentence):
    """
    把字符转为标准的字符，处理编码问题
    :return: 标准化后的字符串
    """
    if not isinstance(sentence, str):
        try:
            sentence = sentence.decode('utf-8')
        except UnicodeDecodeError:
            sentence = sentence.decode('gbk', 'ignore')
    return sentence


def merge_group_sum(arr: List[Dict[str, any]]) -> List[Dict[str, any]]:
    """
    把dict的按指定keygroup，然后count指定字段
    :param arr:
    :return:
    """

    def _group(ms: List[Dict[str, any]], *keys: str) -> List[List[Dict[str, any]]]:
        def _k(m: Dict[str, any], *keys: str) -> str:
            ks = [m[k] for k in keys]
            return "`".join(ks)

        _gkm: Dict[str, List[Dict[str, any]]] = {}  #
        for m in ms:
            k = _k(m, *keys)
            if k not in _gkm:
                _gkm[k] = []
            _gkm[k].append(m)

        return list(_gkm.values())

    def _sum(gm: List[List[Dict[str, any]]], key: str) -> List[Dict[str, any]]:
        retu = []
        for ms in gm:
            first = ms[0]
            for other in ms[1:]:
                ov = other[key]
                first[key] += ov
            retu.append(first)
        return retu

    gm = _group(arr, "keyword", "language_type")
    mc = _sum(gm, "count_num")

    return mc


# 计算编辑距离
# def edit_distance(str1: str, str2: str):
#     len1 = len(str1)
#     len2 = len(str2)
#     dp = numpy.zeros((len1 + 1, len2 + 1))
#     for i in range(len1 + 1):
#         dp[i][0] = i
#     for j in range(len2 + 1):
#         dp[0][j] = j
#
#     for i in range(1, len1 + 1):
#         for j in range(1, len2 + 1):
#             if str1[i - 1] == str2[j - 1]:
#                 temp = 0
#             else:
#                 temp = 1
#             dp[i][j] = min(dp[i - 1][j - 1] + temp, min(dp[i - 1][j] + 1, dp[i][j - 1] + 1))
#     return dp[len1][len2]


# 根据编辑距离计算相似度
# def ed_simility(str1: str, str2: str) -> float:
#     res = edit_distance(str1, str2)
#     max_len = max(len(str1), len(str2))
#     print(res, max_len)
#     return 1 - res / max_len

# 上述编辑距离对比字符串相似度有缺陷  'ab','bc'的相似度为:0.5
# 计算莱文斯坦比。计算公式  r = (sum – ldist) / sum, 其中sum是指str1 和 str2 字串的长度总和，ldist是类编辑距离。注意这里是类编辑距离，在类编辑距离中删除、插入依然+1，但是替换+2。
# def str_similarity(str1: str, str2: str) -> float:
#     import Levenshtein
#     return Levenshtein.ratio(str1, str2)


# def text_to_words(text):
#     import jieba
#     ws = []
#     ss = re.split(r'\s+', text)
#     for s in ss:
#         s = jieba.strdecode(s)  # 统一的转为unicode编码
#         if _is_chinese_chars(s):
#             words = jieba.lcut(s)
#             ws.extend(words)
#         else:
#             ws.append(s)
#     return ws


def _is_chinese_chars(text):
    for char in text:
        cp = ord(char)
        if _is_chinese_char(cp):
            return True
    return False


def _is_chinese_char(cp):
    """Checks whether CP is the codepoint of a CJK character."""
    # This defines a "chinese character" as anything in the CJK Unicode block:
    #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
    #
    # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
    # despite its name. The modern Korean Hangul alphabet is a different block,
    # as is Japanese Hiragana and Katakana. Those alphabets are used to write
    # space-separated words, so they are not treated specially and handled
    # like the all of the other languages.
    if ((0x4E00 <= cp <= 0x9FFF) or  #
            (0x3400 <= cp <= 0x4DBF) or  #
            (0x20000 <= cp <= 0x2A6DF) or  #
            (0x2A700 <= cp <= 0x2B73F) or  #
            (0x2B740 <= cp <= 0x2B81F) or  #
            (0x2B820 <= cp <= 0x2CEAF) or
            (0xF900 <= cp <= 0xFAFF) or  #
            (0x2F800 <= cp <= 0x2FA1F)):  #
        return True

    return False


class DotDict(dict):
    # a.b
    def __getattr__(self, k):
        return self[k]

    def set(self, *ks: any) -> NoReturn:
        """
        [:-1]:string
        [-1]:any
        :param ks:
        """
        if len(ks) <= 1:
            return
        v = ks[-1]
        d = self
        for k in ks[:-2]:
            k = str(k)
            if k not in d:
                d[k] = DotDict()
            d = d[k]
        d[ks[-2]] = v

    # # a.b = c
    # def __setattr__(self, k, v):
    #     # print(f"__setattr__({k},{v})")
    #     self[k] = v

    # # a[b]
    # def __getitem__(self, k):
    #     # print(f"__getitem__({k})")
    #     if not dict.__contains__(self, k):
    #         dict.__setitem__(self, k, DottableDict())
    #     return dict.__getitem__(self, k)
    #
    # def __bool__(self):
    #     return len(self) > 0


def _test_dottabledict():
    c = DotDict()
    c.set("a", "b", "c", "v")
    print(c.a.b)


# def _test_str_similar():
#     st1 = "如何做好行政工作"
#     st2 = "行政工作怎么才能做好呢。"
#     print(str_similarity(st1, st2))  # 0.4


if __name__ == '__main__':
    # _test_str_similar()
    _test_dottabledict()
